package com.adityayadav.parallelism.actors;

import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import com.adityayadav.parallelism.core.ThreadPool;

public class Mailbox<T> {

	protected BlockingQueue<T> messageQueue = new LinkedBlockingQueue<T>();
	protected ThreadPool actorThreadPool;
	protected int actorCount;
	protected IProcessor actor;
	protected String mailboxUUID;
	protected PostOffice postOffice;
	
	public Mailbox(ThreadPool actorThreadPool, PostOffice postOffice, StatefulProcessor actor){
		this.postOffice = postOffice;
		this.mailboxUUID = UUID.randomUUID().toString();
		this.actorCount=1;	
		this.actorThreadPool = actorThreadPool;
		this.actor = actor;
		actor.setMailbox(this);
		actorThreadPool.addTask(actor);// actor is a singleton add it to processing queue
		postOffice.addMailbox(mailboxUUID, this);
	}
	
	public Mailbox(ThreadPool actorThreadPool, PostOffice postOffice, StatelessInfiniteProcessor actor){
		this.postOffice = postOffice;
		this.mailboxUUID = UUID.randomUUID().toString();
		this.actorThreadPool = actorThreadPool;
		this.actor = actor;// infinite actors created and added when message recieved
		postOffice.addMailbox(mailboxUUID, this);
	}
	
	public Mailbox(ThreadPool actorThreadPool, PostOffice postOffice, StatelessFiniteProcessor actor, int actorCount){
		this.postOffice = postOffice;
		this.mailboxUUID = UUID.randomUUID().toString();
		this.actorThreadPool = actorThreadPool;
		this.actor = actor;
		this.actorCount = actorCount;
		for (int i=0; i<actorCount; i++){
			StatelessFiniteProcessor finiteActor = (StatelessFiniteProcessor) actor.clone();
			finiteActor.setMailbox(this);
			actorThreadPool.addTask(finiteActor);//finite actors added rightaway
		}
		postOffice.addMailbox(mailboxUUID, this);
	}
	
	public boolean add(T message){
		boolean success = messageQueue.add(message);
		if (actor instanceof StatelessInfiniteProcessor){
			StatelessInfiniteProcessor infiniteActor = (StatelessInfiniteProcessor) ((StatelessInfiniteProcessor) this.actor).clone();
			infiniteActor.setMailbox(this);
			actorThreadPool.addTask(infiniteActor);
		}
		return success;
	}
	
	public T take() throws Exception {
		return messageQueue.take();
	}
	
	public T peek() throws Exception {
		return messageQueue.peek();
	}
	
	public String getMailboxUUID() {
		return mailboxUUID;
	}

	public PostOffice getPostOffice() {
		return postOffice;
	}
	
	public boolean isTaskQueueEmpty(){
		return messageQueue.isEmpty();
	}
	
	public boolean isIdle(boolean ignorePendingTasks){
		if (!ignorePendingTasks && !messageQueue.isEmpty()) return false;
		return actorThreadPool.isIdle(true);
	}

}
